package sg.edu.nus.iss.stockex;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedList;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;

/**
 * MovingAverageReducer
 */
public class MovingAverageReducer extends Reducer<CodeDateKey, TimestampPriceDataPoint, Text, Text> {

	static long DAY_IN_MS = 24 * 60 * 60 * 1000;

	@Override
	public void reduce(CodeDateKey key, Iterable<TimestampPriceDataPoint> values,
			Context context) throws IOException, InterruptedException {

		TimestampPriceDataPoint nextPoint;
		double sumClose = 0;
		double sumOpen= 0;
		double sumHigh= 0;
		double sumLow= 0;
		double movingAverageClose = 0;
		double movingAverageOpen = 0;
		double movingAverageLow = 0;
		double movingAverageHigh = 0;
		
		int windowSizeInDays = context.getConfiguration().getInt("windowSize", CONFIG.WINDOW_SIZE);
		int windowStepSizeInDays = context.getConfiguration().getInt("windowStepSize", 1);

		long windowSizeInMS = windowSizeInDays * DAY_IN_MS;
		long windowStepSizeInMS = windowStepSizeInDays * DAY_IN_MS;

		Text outKey = new Text();
		Text outVal = new Text();

		SlidingWindow slidingWindow = new SlidingWindow(windowSizeInMS,
				windowStepSizeInMS, DAY_IN_MS);

		Iterator<TimestampPriceDataPoint> valueIterator = values.iterator();
		while (valueIterator.hasNext()) {

			while (!slidingWindow.WindowIsFull() && valueIterator.hasNext()) {

				nextPoint = valueIterator.next();
				System.out.println("#### " + nextPoint.getDate());
				TimestampPriceDataPoint pointCopy = new TimestampPriceDataPoint();
				pointCopy.copy(nextPoint);

				try {
					slidingWindow.AddPoint(pointCopy);
				} catch (Exception e) {
					e.printStackTrace();
				}

			}

			if (slidingWindow.WindowIsFull()) {

			
				LinkedList<TimestampPriceDataPoint> window = slidingWindow
				.GetCurrentWindow();

				// ---------- compute the moving average here -----------
				
				sumOpen = 0;
				sumClose = 0;
				sumHigh = 0;
				sumLow = 0;
				
				
				for (TimestampPriceDataPoint dataPoint: window){
					sumClose += dataPoint.getClosePrice();
					sumOpen += dataPoint.getOpenPrice();
					sumHigh += dataPoint.getHighPrice();
					sumLow += dataPoint.getLowPrice();
				}

				movingAverageClose = sumClose/window.size();
				movingAverageOpen = sumOpen/window.size();
				movingAverageHigh = sumHigh/window.size();
				movingAverageLow = sumLow/window.size();
				
				String strBackDate = window.getLast().getDate();
				
				//outKey.set("Code: " + key.getCode() + ", Date: "
				//		+ strBackDate);
				//outVal.set("Moving Average: " + movingAverage);

				outKey.set(key.getCode() + "\t" + strBackDate);
				outVal.set(movingAverageHigh + "\t" + movingAverageLow + "\t" + movingAverageClose + "\t" + movingAverageOpen );

				context.write(outKey, outVal);

				//step window forward

				slidingWindow.slideWindowForward();

			}

		} // end while

	} // end reduce

}
